package room_event

import (
	"encoding/json"
	"github.com/spf13/cast"
	"longmen/server/config/global"
	"longmen/server/internal/rpc/tencentim/cache"
	"longmen/server/internal/rpc/tencentim/chat"
	"longmen/server/internal/rpc/tencentim/imsdk"
	messages2 "longmen/server/internal/rpc/tencentim/messages"
	"longmen/server/internal/rpc/tencentim/utils"
	"longmen/server/internal/services/consumers/publisher"
	"longmen/server/pkg/db/mysql/models/models_main"
	pb "longmen/server/pkg/protobuf"
)

var watchStatChan = make(chan watchStatPayload, 10000)

func handlingPingMessage(c *chat.ClientUser, ip string, dataByte []byte) (ret *pb.CallbackResp, errRes error) {

	ret = &pb.CallbackResp{
		ActionStatus: imsdk.OKActionStatus,
		ErrorCode:    imsdk.TIM_CODE_OK,
	}
	rep := &messages2.ReplyMessage{
		Type:    messages2.PongMessageType,
		Code:    0,
		Message: "",
	}
	IncrIpStats(c.RoomId, ip)
	WatchTimeStat(c.RoomId, c.UserId, dataByte)
	return utils.CallbackOkAndExt(rep, cast.ToString(c.RoomId)), nil //返回改过的rep
}

func IncrIpStats(roomId int, clientIp string) {
	b, err := global.BloomExists(global.BLOOM_ANCHOR, cast.ToString(roomId))
	if err != nil {
		global.Log.Errorf("IncrIpStats BloomExists err:%v", err)
		return
	}
	if b {
		heat := cache.NewRoomHeat(roomId)
		heat.IncrIpCount(clientIp)
		//cache.IncGlobalIp(clientIp) 改到getCommonConfig接口统计
	}
}

//func WatchTimeStat(roomId int, userId int, b []byte) {
//	go func() {
//		d := struct {
//			DeviceId string `json:"device_id"`
//		}{}
//		err := json.Unmarshal(b, &d)
//		if err != nil {
//			global.Log.Errorf("WatchTimeStat jsonUnmarshal err:%v", err)
//			return
//		}
//
//		err = global.PublisherActivity.Send(publisher.LiveWatchStats, struct {
//			RoomId int
//			models_main.User
//		}{
//			RoomId: roomId,
//			User: models_main.User{
//				Id:         userId,
//				DeviceInfo: d.DeviceId,
//			},
//		})
//		if err != nil {
//			global.Log.Errorf("publish chanel %s", err.Error())
//		}
//	}()
//}

type watchStatPayload struct {
	RoomId   int
	UserId   int
	DeviceId string
}

func InitWatchTimeStatWorkerPool(workerNum int) {
	for i := 0; i < workerNum; i++ {
		go func() {
			for payload := range watchStatChan {
				err := global.PublisherActivity.Send(publisher.LiveWatchStats, struct {
					RoomId int
					models_main.User
				}{
					RoomId: payload.RoomId,
					User: models_main.User{
						Id:         payload.UserId,
						DeviceInfo: payload.DeviceId,
					},
				})
				if err != nil {
					global.Log.Errorf("<<<< publish channel error: %v", err)
				}
			}
		}()
	}
}

func WatchTimeStat(roomId int, userId int, b []byte) {
	d := struct {
		DeviceId string `json:"device_id"`
	}{}
	if err := json.Unmarshal(b, &d); err != nil {
		global.Log.Errorf("WatchTimeStat jsonUnmarshal err:%v", err)
		return
	}

	select {
	case watchStatChan <- watchStatPayload{RoomId: roomId, UserId: userId, DeviceId: d.DeviceId}:
	default:
		global.Log.Warnf("watchStatChan full, dropped roomId=%d userId=%d", roomId, userId)
	}
}
